package com.lei.mydb.backend.dm.pageCache;


import com.lei.mydb.backend.common.AbstractCache;
import com.lei.mydb.backend.dm.page.Page;
import com.lei.mydb.backend.dm.page.PageImpl;
import com.lei.mydb.backend.utils.Panic;

import java.io.IOException;
import java.io.RandomAccessFile;
import java.nio.ByteBuffer;
import java.nio.channels.FileChannel;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
import com.lei.mydb.common.Error;

/**
 * @author lbwxxc
 * @date 2025/4/29 21:46
 * @description: 实现类
 */
public class PageCacheImpl extends AbstractCache<Page> implements PageCache {

    private static final int MEM_MIN_LIM = 10;
    public static final String DB_SUFFIX = ".db";

    private RandomAccessFile file;
    private FileChannel fc;
    private Lock fileLock;

    private AtomicInteger pageNumbers;

    PageCacheImpl(RandomAccessFile file, FileChannel fileChannel, int maxResource) {
        super(maxResource);
        if(maxResource < MEM_MIN_LIM) {
            Panic.panic(Error.MemTooSmallException);
        }
        long length = 0;
        try {
            length = file.length();
        } catch (IOException e) {
            Panic.panic(e);
        }
        this.file = file;
        this.fc = fileChannel;
        this.fileLock = new ReentrantLock();
        this.pageNumbers = new AtomicInteger((int) length / PAGE_SIZE);
    }

    @Override
    public void release(Page page) {
        release(page.getPageNumber());
    }

    @Override
    public Page getPage(int pgno) throws Exception {
        // 调用 AbstractCache 的 get 方法，内部也会调用 this.getForCache 方法
        return get(pgno);
    }


    // 异常回复
    @Override
    public void truncateByBgno(int maxPgno) {
        long size = pageOffset(maxPgno + 1);
        try {
            file.setLength(size);
        } catch (IOException e) {
            Panic.panic(e);
        }
        pageNumbers.set(maxPgno);
    }

    @Override
    public int getPageNumber() {
        return pageNumbers.intValue();
    }

    // 关闭所有资源
    @Override
    public void close() {
        super.close();
        try {
            fc.close();
            file.close();
        } catch (IOException e) {
            Panic.panic(e);
        }
    }

    // 写回数据页
    @Override
    public void flushPage(Page pg) {
        this.flush(pg);
    }

    // 创建新的数据页，并写回文件
    @Override
    public int newPage(byte[] initData) {
        int pano = pageNumbers.incrementAndGet();
        PageImpl page = new PageImpl(pano, initData, null);
        flush(page);
        return pano;
    }

    // ========= 重写 AbstractCache<T> 方法 =========
    @Override
    protected Page getForCache(long key) {
        int pgno = (int) key;
        long offset = PageCacheImpl.pageOffset(pgno);
        ByteBuffer buffer = ByteBuffer.allocate(PAGE_SIZE);
        fileLock.lock();
        try {
            fc.position(offset);
            fc.read(buffer);
        } catch (IOException e) {
            Panic.panic(e);
        }
        fileLock.unlock();
        return new PageImpl(pgno, buffer.array(), this);
    }

    @Override
    protected void releaseForCache(Page obj) {
        if (obj.isDirty()) {
            flush(obj);
            obj.setDirty(false);
        }
    }

    // ========= 私有方法 =========

    // 写回数据页
    private void flush(Page pg) {
        int pgno = pg.getPageNumber();
        // pg 的偏移
        long offset = PageCacheImpl.pageOffset(pgno);
        try {
            fileLock.lock();
            ByteBuffer buffer = ByteBuffer.wrap(pg.getData());
            fc.position(offset);
            fc.write(buffer);
            fc.force(false);
        } catch (IOException e) {
            Panic.panic(e);
        }
        fileLock.unlock();
    }

    // 计算数据在文件中的偏移
    private static long pageOffset(int pgno) {
        return (pgno - 1) * PAGE_SIZE;
    }

}
